package core.sql.环境;

import org.apache.spark.rdd.RDD;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * 1、JSON需要进行压缩，展开的可能无法读取。
 * 2、需要配置                    .option("mode", "PERMISSIVE") // 允许坏记录
 * .option("badRecordsPath", "data/bad_records") // 指定错误记录的输出路径
 * 否则可能会报错
 */
public class Spark02_MODEL {

    public static void main(String[] args) {
        SparkSession sparkSession = SparkSession
                .builder()
                .master("local[*]")
                .appName("Spark02_MODEL")
                .getOrCreate();

        try {
            // 读取 JSON 数据
            Dataset<Row> json = sparkSession.read()
                    .option("mode", "PERMISSIVE") // 允许坏记录
                    .option("badRecordsPath", "data/bad_records") // 指定错误记录的输出路径
                    .json("data/user_sql.json");

            json.show();
            // 缓存解析后的结果
            json.cache();

            // 创建临时视图
            json.createOrReplaceTempView("temp_table");

            // 示例：执行 SQL 查询
            Dataset<Row> result = sparkSession.sql("SELECT * FROM temp_table WHERE age > 35");
            result.show(); // 显示结果

//            // 处理错误记录
//            Dataset<Row> corruptRecords = sparkSession.read().json("data/bad_records/*");
//            corruptRecords.show(); // 显示错误记录

        } catch (Exception e) {
            e.printStackTrace(); // 捕获并打印异常
        } finally {
            sparkSession.close();
        }
    }

}

